package com.yujiahao.bigdata.test

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark01_TestHotTop10_Accumulator {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接

    val conf: SparkConf = new SparkConf().setAppName("TestTop10").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑


    //TODO 需求一的问题：优化方案四--使用累加器
    //先读取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    //新建一个累加器对象。
    val acc: My_Accumulator = new My_Accumulator()
    // 向Spark中注册。
    sc.register(acc, "HotCategory")

    // 获取值并累加数据。
    rdd.foreach(
      data => {
        val datas = data.split("_")
        if (datas(6) != "-1") {
          // 点击的场合
          acc.add((datas(6), "click"))
        } else if (datas(8) != "null") {
          // 下单的场合
          val ids = datas(8).split(",")
          ids.foreach(
            id => {
              acc.add((id, "order"))
            }
          )
        } else if (datas(10) != "null") {
          // 支付的场合
          val ids = datas(10).split(",")
          ids.foreach(
            id => {
              acc.add((id, "pay"))
            }
          )
        }
      }
    )
    // 获取值

    acc.value.toList.map(_._2).sortBy(
      num => {
        (num.clickCnt, num.orderCnt, num.payCnt)
      }
    )(Ordering.Tuple3(Ordering.Int.reverse, Ordering.Int.reverse, Ordering.Int.reverse)).take(10).foreach(println)

    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

  //todo 将累加器输出的内容进行封装
  case class HotCategoryClick(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)
  //todo 这是自定义的累加器以及步骤：
  // 1、写一个类继承AccumulatorV2
  // 2、确定泛型
  // 输入泛型：[String,String]输出泛型：[品类（点击数量，下单数量，支付数量）]--一般习惯于将这几个使用样例类封装一下
  // 3、重写6个方法

  class My_Accumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategoryClick]] {
    //既然是返回的集合么有就要先new一个
    private val hcMap = mutable.Map[String, HotCategoryClick]()

    //判断是否清空
    override def isZero: Boolean = {
      hcMap.isEmpty
    }

    //复制
    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategoryClick]] = {
      //复制的不如直接new一个出来
      new My_Accumulator()
    }

    //重置
    override def reset(): Unit = {
      hcMap.clear()
    }

    //添加
    override def add(v: (String, String)): Unit = {
      val (cid, actionType) = v
      //判断集合中是否存在cid
      // {}
      // get
      val obj: HotCategoryClick = hcMap.getOrElse(cid, HotCategoryClick(cid, 0, 0, 0))
      //将品类进行叠加
      actionType match {
        case "click" => {
          obj.clickCnt += 1
        }
        case "order" => {
          obj.orderCnt += 1
        }
        case "pay" => {
          obj.payCnt += 1
        }
      }
      //添加完了之后将添加的东西更新进去
      hcMap.update(cid, obj)
    }

    //合并
    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategoryClick]]): Unit = {
      //这里面就是将多个累加的值进行合并
      val otherMap: mutable.Map[String, HotCategoryClick] = other.value
      otherMap.foreach {
        case (cid, otherHC) => {
          val thisHC = hcMap.getOrElse(cid, HotCategoryClick(cid, 0, 0, 0))
          thisHC.clickCnt += otherHC.clickCnt
          thisHC.orderCnt += otherHC.orderCnt
          thisHC.payCnt += otherHC.payCnt
          hcMap.update(cid, thisHC)
        }
      }
    }

    //返回累加结果
    override def value: mutable.Map[String, HotCategoryClick] = hcMap

  }
}
